package main

import (
	"rpc"
	"http"
	"net"
	"os"
	"log"
	"fmt"
	"tribproto"
	"flag"
	"io"
	"crypto/sha1"
	"json"
	"time"
	//"storageproto"
	"storagerpc"
	"storageserver"
)

type Tribserver struct {
	//Table *hashTbl.HashTable
	ss *storageserver.Storageserver
}

// Structure that keeps track of the Tribbles for a specific subscription.
// Used for GetTribblesBySubscription
type Subscription struct {
	Userid string
	Index int
	CurTrib tribproto.Tribble
	Indices []string
}

// Set up a new Tribserver, and initialize the associated hashtable
func NewTribserver(ss *storageserver.Storageserver) *Tribserver {
	return &Tribserver{ss}
}

// Add a new user to the system
func (ts *Tribserver) CreateUser(args *tribproto.CreateUserArgs, reply *tribproto.CreateUserReply) os.Error {
	// Check to see if the user is already in the system
	_, code := ts.ss.GET (args.Userid)

	// If the user is in the system, change the status
	if code == 0 {
		reply.Status = tribproto.EEXISTS
	// Otherwise, add the username to the hashtable for easy duplicate checks
	} else {
		ts.ss.AddToList(args.Userid, "")
	}

	return nil
}

// Add a subscription for a specific user
func (ts *Tribserver) AddSubscription(args *tribproto.SubscriptionArgs, reply *tribproto.SubscriptionReply) os.Error {
	// Check to see if the specified user exists
	_, code := ts.ss.GET (args.Userid)

	// If it doesn't, change the status and quit
	if code != 0 {
		reply.Status = tribproto.ENOSUCHUSER
		return nil
	}

	// Check to see if the target user exists
	_, code = ts.ss.GET (args.Targetuser)

	// If it doesn't, change the status and quit
	if code != 0 {
		reply.Status = tribproto.ENOSUCHTARGETUSER
		return nil
	}

	// Check to see if the target user is already subscribed to the target user

	// Get the list of subscriptions for the user
	subs, _ := ts.ss.GETLIST (fmt.Sprintf("%s:subscriptions", args.Userid))

	if subs != nil {
		// Check each subscription for a match to the target user
		for i := 0; i < len(subs); i++ {
			// If they match, do nothing else and quit the function
			if subs[i] == args.Targetuser {
				reply.Status = tribproto.EEXISTS
				return nil
			}
		}
	}

	// If the target user is not already subscribed, add the user to the list
	ts.ss.AddToList(fmt.Sprintf("%s:subscriptions", args.Userid), args.Targetuser)

	return nil
}

// Remove a subscription from a user's list of subscriptions
func (ts *Tribserver) RemoveSubscription(args *tribproto.SubscriptionArgs, reply *tribproto.SubscriptionReply) os.Error {
	// Check to see if the passed in arguments are valid users, and change the reply on error
        _, code := ts.ss.GET (args.Userid)

        if code != 0 {
		reply.Status = tribproto.ENOSUCHUSER
                return nil
        }

        _, code = ts.ss.GET (args.Targetuser)

        if code != 0 {
		reply.Status = tribproto.ENOSUCHTARGETUSER
                return nil
        }

	// Remove the target user from the list of subscriptions
        ts.ss.RemoveFromList(fmt.Sprintf("%s:subscriptions", args.Userid), args.Targetuser)

	return nil
}

// Return a list of subscriptions for a specific user
func (ts *Tribserver) GetSubscriptions(args *tribproto.GetSubscriptionsArgs, reply *tribproto.GetSubscriptionsReply) os.Error {
	// Check to see if the passed in argument is a valid user, and change the reply on error
        _, code := ts.ss.GET (args.Userid)

        if code != 0 {
                reply.Status = tribproto.ENOSUCHUSER
                return nil
        }

	// Get the list of subscriptions for a specific user
	reply.Userids, _ = ts.ss.GETLIST (fmt.Sprintf("%s:subscriptions", args.Userid))

	return nil
}

// Add a tribble to the hashtable
func (ts *Tribserver) PostTribble(args *tribproto.PostTribbleArgs, reply *tribproto.PostTribbleReply) os.Error {
	// Check to see if the passed in argument is a valid user, and change the reply on error
	_, code := ts.ss.GET (args.Userid)

	if code != 0 {
		reply.Status = tribproto.ENOSUCHUSER
		return nil
	}

	// Initialize a new Tribble for the new post
	var trib tribproto.Tribble
	trib.Userid = args.Userid
	trib.Posted = time.Nanoseconds()
	trib.Contents = args.Contents

	// Convert the structure into a string using Marshalling
	str, _ := json.Marshal(trib)

	// The tribble will be indexed by the hash of the tribble structure's string
        hashval := sha1.New()
        io.WriteString(hashval, string(str))
        hashstring := fmt.Sprintf("%x", hashval.Sum())

	// Add the tribble's index to the user's tribble index list
        ts.ss.AddToList(fmt.Sprintf("%s:tribindex", args.Userid), hashstring)

	// Add the tribble to the hashtable
	ts.ss.PUT(fmt.Sprintf("%s:%s", args.Userid, hashstring), string(str))
	
	return nil
}

// Get a list of the most recent 100 tribbles
func (ts *Tribserver) GetTribbles(args *tribproto.GetTribblesArgs, reply *tribproto.GetTribblesReply) os.Error {
	// Check to see if the passed in argument is a valid user, and change the reply on error
        _, code := ts.ss.GET (args.Userid)

        if code != 0 {
                reply.Status = tribproto.ENOSUCHUSER
                return nil
        }
	
	// Get the tribble indices for the user
        indices, _ := ts.ss.GETLIST(fmt.Sprintf("%s:tribindex", args.Userid))
	if indices == nil {
		return nil
	}

	// Initialize the value of the returned tribbles
	reply.Tribbles = make([]tribproto.Tribble,1,100)
	
	// Cycle through the tribbles indicated by the tribble index and add them to the reply structure
	for i := len(indices)-1; i >= 0; i-- {
		// Get the tribble indicated by the index
		entry, code := ts.ss.GET(fmt.Sprintf("%s:%s", args.Userid, indices[i]))
		if code != 0 {
			return nil
		}

		var trib tribproto.Tribble

		// Append the tribble to the end of the reply structure
		if entry != "" {
			json.Unmarshal([]byte(entry), &trib)
			if i == len(indices)-1 {
				reply.Tribbles[0] = trib
			} else {
				reply.Tribbles = append (reply.Tribbles, trib)
			}
		}
	}

	return nil
}

// Get a list of the most recent 100 tribbles from all the users a user is subscribed to
func (ts *Tribserver) GetTribblesBySubscription(args *tribproto.GetTribblesArgs, reply *tribproto.GetTribblesReply) os.Error {
	// Check to see if the passed in argument is a valid user, and change the reply on error
        _, code := ts.ss.GET (args.Userid)

        if code != 0 {
                reply.Status = tribproto.ENOSUCHUSER
                return nil
	}

	// Get the subscriptions for a specific user
	sub_string, _ := ts.ss.GETLIST(fmt.Sprintf("%s:subscriptions", args.Userid))

	// Allocate Subscriptions array
	subs := make([]Subscription, len(sub_string))

	// Initialize the Subscriptions array with the information from each subscriptions
	for i := 0; i < len(subs); i++ {
		// Get the indices for the tribbles of the subscription
		subs[i].Indices, _ = ts.ss.GETLIST(fmt.Sprintf("%s:tribindex", sub_string[i]))
		// The index of the current subscription
		subs[i].Index = len(subs[i].Indices)
		// The userid of the target subscription
		subs[i].Userid = sub_string[i]
		// Get the value of the current (most recent) tribble of the subscription
		if subs[i].Index > 0 {
			subs[i].Index --
			// Retrieve the value of the subscription
			trib_marsh, code := ts.ss.GET(fmt.Sprintf("%s:%s", subs[i].Userid, subs[i].Indices[subs[i].Index]))
			if code != 0 {
				return nil
			}
			// Unmarshal the strng into a Tribble structure
			json.Unmarshal([]byte(trib_marsh), &subs[i].CurTrib)
		}
	}

	// Allocate the reply's tribble structure
        reply.Tribbles = make([]tribproto.Tribble, 1, 100)

	var minInd int

	// For up to 100 tribbles, get the tribbles in reverse chronological order
        for i := 0; i < 100; i++ {
		minInd = -1
		// Find the most recent tribbles out of all the subscriptions
		for j := 0; j < len(subs); j++ {
			if (minInd == -1 && subs[j].Index > -1) ||
				(minInd > -1 && subs[j].Index > -1 && subs[j].CurTrib.Posted > subs[minInd].CurTrib.Posted) {
				minInd = j
			}
		}

		// If a minimum tribble was found, append the tribble
		if minInd > -1 {
			// Append the tribble to the reply tribble array
			if i == 0 {
				reply.Tribbles[0] = subs[minInd].CurTrib
			} else {
				reply.Tribbles = append(reply.Tribbles, subs[minInd].CurTrib)
			}
			
			// Update the current tribble for the subscription
			subs[minInd].Index --
			if subs[minInd].Index > -1 {
				trib_marsh, code := ts.ss.GET(fmt.Sprintf("%s:%s", subs[minInd].Userid, subs[minInd].Indices[subs[minInd].Index]))
				if code != 0 {
					return nil                                                                                                     
				}
				json.Unmarshal([]byte(trib_marsh), &subs[minInd].CurTrib)
			}
		// Otherwise, there aren't any other tribbles to add
		} else {
			break
		}
        }

	return nil
}

var portnum *int = flag.Int("port", 9009, "port # to listen on")
var storageMasterNodePort *string = flag.String("master", "", "Storage master node.  Defaults to its own port.")
var numNodes *int = flag.Int("N", 0, "Become the master.  Specifies the number of nodes in the system, including the master.")
var nodeID *uint = flag.Uint("id", 0, "The node ID to use for consistent hashing.  Should be a 32 bit number.")


func main() {
	flag.Parse()
	if (*storageMasterNodePort == "") {
		// Single node execution
		*storageMasterNodePort = fmt.Sprintf("localhost:%d", *portnum)
		if (*numNodes == 0) {
			*numNodes = 1
			log.Println("Self-mastering, setting nodes to 1")
		}
	}
	log.Printf("Server starting on port %d\n", *portnum);
	ss := storageserver.NewStorageserver(*storageMasterNodePort,*numNodes, *portnum,uint32(*nodeID))
	ts := NewTribserver(ss)
	rpc.Register(ts)
	srpc := storagerpc.NewStorageRPC(ss)
	rpc.Register(srpc)
	rpc.HandleHTTP()
	l, e := net.Listen("tcp", fmt.Sprintf(":%d", *portnum))
	if e != nil {
		log.Fatal("listen error:", e)
	}
	http.Serve(l,nil)
}

/*
func main() {
	flag.Parse()
	log.Printf("Server starting on port %d\n", *portnum);
	ts := NewTribserver()
	rpc.Register(ts)
	rpc.HandleHTTP()
	l, e := net.Listen("tcp", fmt.Sprintf(":%d", *portnum))
	if e != nil {
		log.Fatal("listen error:", e)
	}
	http.Serve(l, nil)
}
*/
